#ifndef __VAP_BASE__
#define __VAP_BASE__

/**
 * vap_base 视频分析流水线的实现
 * 设计模式：订阅-观察者模式
 * 特点：处理节点异步执行，互补干扰
 * 
*/

#include <list>
#include <mutex>
#include <memory>
#include <queue>

#include "vap_type.h"
#include "vap_utils.h"
namespace vap_base {

    /**
     * 类型：订阅者
    */
    class vap_subscriber
    {
    private:

    public:
        vap_subscriber(){};
        ~vap_subscriber(){};
        // 删除拷贝赋值和拷贝构造函数，禁止子类调用拷贝
        vap_subscriber(const vap_subscriber&)=delete;
        vap_subscriber& operator=(const vap_subscriber&) =delete;

        // 声明成了纯虚函数，需要由子类进行实现，涉及到具体的
        virtual void meta_flow(std::shared_ptr<vap_type::vap_meta> meta) = 0 ;
    };

    /**
     * 类型：发布者
    */
    class vap_publisher
    {
    protected:
        std::mutex subscriber_mutex_;
        std::list<std::shared_ptr<vap_subscriber>> subscribers_; //订阅者列表
    public:
        void add_subscriber(std::shared_ptr<vap_subscriber> subscriber)
        {
            std::lock_guard<std::mutex> lock(subscriber_mutex_);
            subscribers_.push_back(subscriber);
        }

        void remove_subscriber(std::shared_ptr<vap_subscriber> subscriber)
        {
            std::lock_guard<std::mutex> lock(subscriber_mutex_);
            subscribers_.remove_if(subscriber);
        }

        // 将meta信息推送到订阅者的缓存中。这里声明成了虚函数，说明子类可以重写
        virtual void push_meta(std::shared_ptr<vap_type::vap_meta> meta) {
            std::lock_guard<std::mutex> lock(subscriber_mutex_);
            for(auto it : subscribers_)
                it->meta_flow(meta);
        }
    };

    enum vap_node_type
    {
        SRC,    // 数据输入节点
        DES,    // 数据输出节点
        MID     // 数据中间处理节点
    };
    /**
     * 类型：vap_node
     * 描述：流水线中的基本处理单元，负责每个节点上的处理认为
    */
    class vap_node: public vap_publisher,
                    public vap_subscriber,
                    public std::enable_shared_from_this<vap_node>
    {
    private:
        vap_node_type type_;    //节点类型
        std::string node_name_;     //节点名称
        std::thread handle_run_t;       // 处理线程号
        std::thread dispatch_run_t;     //后处理线程号
        std::list<std::shared_ptr<vap_node>> pre_nodes_; //订阅者列表
    protected:
        std::queue<std::shared_ptr<vap_type::vap_meta>> in_queue_; //节点处理操作前的缓存
        std::mutex in_queue_lock_;
        std::queue<std::shared_ptr<vap_type::vap_meta>> out_queue_; // 节点处理操作后的缓存
        vap_utils::vap_semaphore in_queue_semaphore_;       //in_queue_的信号量
        vap_utils::vap_semaphore out_queue_semaphore_;      //out_queue_的信号量
        
    public:
        // 声明成纯虚函数，希望由子节点根据具体的vap_node_type进行实现
        virtual void handle_run() =0; // 节点处理操作，从in_queue_取meta 处理，处理完后的meta搬运至out_queue_
        virtual void dispatch_run() =0; // 搬运out_queue_到下一个节点处
        virtual std::shared_ptr<vap_type::vap_meta> handle_frame_meta(std::shared_ptr<vap_type::vap_frame> frame_meta){return frame_meta;}
        virtual std::shared_ptr<vap_type::vap_meta> handle_control_meta(std::shared_ptr<vap_type::vap_control> control_meta){return control_meta;}
        virtual void  handle_frame_meta(const std::vector<std::shared_ptr<vap_type::vap_frame>>& frame_meta_with_batch) {}

        // 和前面的节点分离
        void detach()
        {
            for(auto it : pre_nodes_)
            {
                it->remove_subscriber(shared_from_this());  
            }
            this->pre_nodes_.clear();
        };

        // 连接节点
        void attach_to(std::vector<std::shared_ptr<vap_node>> pre_nodes){
            for(auto it : pre_nodes)
            {
                it->add_subscriber(shared_from_this());
                this->pre_nodes_.push_back(it);
            }
        };

        vap_node(std::string node_name,vap_node_type type):node_name_(node_name),type_(type)
        {
            initialized();
        };
        ~vap_node(){}

        void initialized()
        {
            this->handle_run_t = std::thread(&handle_run,this);
            this->dispatch_run_t = std::thread(&dispatch_run,this);
        };

        virtual std::string get_name(){return this->node_name_;}

        virtual void meta_flow(std::shared_ptr<vap_type::vap_meta> meta) final 
        {
            if(meta == nullptr)
                return;
            std::lock_guard<std::mutex> lock(this->in_queue_lock_);
            this->in_queue_.push(meta);
            this->in_queue_semaphore_.signal();
        }
    };

    /**
     * 类型：vap_src_node
     * 描述：视频分析流水线的输入节点
    */
    class vap_src_node: public vap_node
    {
    protected:
        // 具体的实现细节需要子类进行重写
        virtual void handle_run() override 
        {
            throw std::runtime_error("must have re-implementaion for 'handle_run' method in src nodes!");
        }
        virtual std::shared_ptr<vap_type::vap_meta> handle_frame_meta(std::shared_ptr<vap_type::vap_frame> frame_meta) override final 
        {
            throw std::runtime_error("'handle_frame_meta' method could not be called in src nodes!");
        }
        virtual std::shared_ptr<vap_type::vap_meta> handle_control_meta(std::shared_ptr<vap_type::vap_control> control_meta) override final 
        {
            throw std::runtime_error("'handle_control_meta' method could not be called in src nodes!");
        }
        virtual void  handle_frame_meta(const std::vector<std::shared_ptr<vap_type::vap_frame>>& frame_meta_with_batch) override final 
        {
            throw std::runtime_error("'handle_frame_meta' method could not be called in src nodes!");
        }
        virtual void dispatch_run() override final
        {
            while(true)
            {
                this->out_queue_semaphore_.wait();
                auto meta = out_queue_.front();
                this->push_meta(meta);
                out_queue_.pop();
            }
        }

        vap_src_node(std::string node_name, int channel_id, float resize_ratio = 1.0):\ 
                vap_node(node_name,vap_base::vap_node_type::SRC),channel_id_(channel_id),resize_ratio_(resize_ratio)\
                {assert(resize_ratio_ > 0 && resize_ratio_ <= 1.0f);}
        int original_fps_ = -1;
        int original_width_ = 0;
        int original_height_ = 0;
        int channel_id_;
        float resize_ratio_;
        vap_utils::vap_gate gate;

    public:
        ~vap_src_node(){};
        void start(){this->gate.open();};
        void stop(){this->gate.close();};
        void speak(){};
        int get_original_fps() const {return this->original_fps_;};
        int get_original_width() const {return this->original_width_;};
        int get_original_height() const {return this->original_height_;};
    };

    /**
     * 类型：vap_des_node
     * 描述：视频分析流水线的输出节点
    */
    class vap_des_node: public vap_node
    {
    protected:
        int channel_id_;
    public:
        virtual void handle_run() override final
        {
            while(true)
            {
                this->in_queue_semaphore_.wait();
                auto in_meta = this->in_queue_.front();
                std::shared_ptr<vap_type::vap_meta> out_meta;
                if(in_meta->get_type() == vap_type::vap_meta_type::control)
                {
                    auto meta_2_handle = std::dynamic_pointer_cast<vap_type::vap_control>(in_meta);
                    out_meta = this->handle_control_meta(meta_2_handle);
                }
                else if(in_meta->get_type() == vap_type::vap_meta_type::frmae)
                {
                    auto meta_2_handle = std::dynamic_pointer_cast<vap_type::vap_frame>(in_meta);
                    out_meta = this->handle_frame_meta(meta_2_handle);
                }
                else {
                    throw std::runtime_error("invalid meta type!");
                }
                this->in_queue_.pop();
            }
        }
        virtual void dispatch_run() override final 
        {
            throw std::runtime_error("do nothing for 'dispatch_run' method in des nodes!");// 因为是最后一个节点，因此什么都不需要做
        };  
        virtual std::shared_ptr<vap_type::vap_meta> handle_frame_meta(std::shared_ptr<vap_type::vap_frame> frame_meta) override 
        {
            throw std::runtime_error("must have re-implementaion for 'handle_frame_meta' method in des nodes!"); // 具体保存数据的方法
        }
        virtual std::shared_ptr<vap_type::vap_meta> handle_control_meta(std::shared_ptr<vap_type::vap_control> control_meta) override 
        {
            throw std::runtime_error("must have re-implementaion for 'handle_control_meta' method in des nodes!"); // 具体处理control数据的方法
        }
        vap_des_node(std::string node_name,int channel_id): \ 
                vap_node(node_name,vap_base::vap_node_type::DES),channel_id_(channel_id){};
        ~vap_des_node(){};
    };

    /**
     * 类型：vap_infer_node
     * 描述：视频分析流水线的中间推理节点,这里的抽象是基于昇腾的边缘框架，例如Atlas200等，当然也可以使用其他的推理框架
     * 
    */
    class vap_infer_node: public vap_node
    {
    protected:
        int frame_meta_handle_batch_;
        std::vector<std::shared_ptr<vap_type::vap_frame>> frame_meta_batch_cache_;
    public:
        vap_infer_node(std::string node_name,int frame_meta_handle_batch=1):vap_node(node_name,vap_base::vap_node_type::MID), \ 
                    frame_meta_handle_batch_(frame_meta_handle_batch){};
        ~vap_infer_node(){};

        virtual void preprocess(const std::vector<std::shared_ptr<vap_type::vap_frame>>& frame_meta_with_batch, std::vector<cv::Mat>& mats_to_infer) = 0;
        virtual void infer(std::vector<cv::Mat>& mats_to_infer, std::vector<cv::Mat>& raw_outputs) = 0;
        virtual void postprocess(const std::vector<cv::Mat>& raw_outputs, const std::vector<std::shared_ptr<vap_type::vap_frame>>& frame_meta_with_batch) = 0;

        void infer_combinations_time_cost(int data_size, int preprocess_time, int infer_time, int postprocess_time) 
        {
            std::stringstream sstream;
            sstream << "\n########## infer combinations summary ##########\n";
            sstream << " node_name:" << this->get_name() << "\n";
            sstream << " data_size:" << data_size << "\n";
            sstream << " preprocess_time:" << preprocess_time << "ms\n";
            sstream << " infer_time:" << infer_time << "ms\n";
            sstream << " postprocess_time:" << postprocess_time << "ms\n";
            sstream << "########## infer combinations summary ##########\n";    

            VAP_LOG_DEBUG("%s\n",sstream.str());
        }

        void run_infer_combinations(const std::vector<std::shared_ptr<vap_type::vap_frame>>& frame_meta_with_batch)
        {
            std::vector<cv::Mat> mats_to_infer;
            std::vector<cv::Mat> raw_outputs;


            auto start_time = std::chrono::system_clock::now();
            preprocess(frame_meta_with_batch,mats_to_infer);
            auto preprocess_time = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now() - start_time);

            if(mats_to_infer.size() == 0)
                return;

            auto start_time = std::chrono::system_clock::now();
            infer(mats_to_infer,raw_outputs);
            auto infer_time = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now() - start_time);

            auto start_time = std::chrono::system_clock::now();
            postprocess(mats_to_infer,frame_meta_with_batch);
            auto postprocess_time = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now() - start_time);

            infer_combinations_time_cost(mats_to_infer.size(),preprocess_time.count(),infer_time.count(),postprocess_time.count());
        };

        virtual std::shared_ptr<vap_type::vap_meta> handle_frame_meta(std::shared_ptr<vap_type::vap_frame> frame_meta) override final
        {
            std::vector<std::shared_ptr<vap_type::vap_frame>> frame_meta_with_batch {frame_meta};
            run_infer_combinations(frame_meta_with_batch);    
            return frame_meta;
        }; 

        virtual void  handle_frame_meta(const std::vector<std::shared_ptr<vap_type::vap_frame>>& frame_meta_with_batch) override final
        {
            const auto& frame_meta_with_batch_ = frame_meta_with_batch;
            run_infer_combinations(frame_meta_with_batch_);    
        }; 

        virtual std::shared_ptr<vap_type::vap_meta> handle_control_meta(std::shared_ptr<vap_type::vap_control> control_meta) override 
        {
            throw std::runtime_error("must have re-implementaion for 'handle_control_meta' method in des nodes!"); // 具体处理control数据的方法
        }

        virtual void handle_run() override 
        {
            std::shared_ptr<vap_type::vap_meta> out_meta;

            while (true)
            {
                this->in_queue_semaphore_.wait();
                auto in_meta = in_queue_.front();

                if(in_meta->get_type() == vap_type::frmae)
                {
                    auto meta_2_handle = std::dynamic_pointer_cast<vap_type::vap_frame>(in_meta);
                    // 一帧一帧处理
                    if(frame_meta_handle_batch_ == 1)
                    {
                        out_meta = this->handle_frame_meta(meta_2_handle);
                        if(out_meta !=nullptr )
                        {
                            this->out_queue_.push(out_meta);
                            this->out_queue_semaphore_.signal();
                        }
                    }
                    else  // 以batch 的方式处理
                    {
                        frame_meta_batch_cache_.push_back(meta_2_handle);
                        if(frame_meta_batch_cache_.size() >= frame_meta_handle_batch_)
                        {
                            this->handle_frame_meta(frame_meta_batch_cache_);
                            for(auto it : frame_meta_batch_cache_)
                            {
                                this->out_queue_.push(it);
                            }
                            this->out_queue_semaphore_.signal();
                            frame_meta_batch_cache_.clear();
                        }
                    }
                }
                else if(in_meta->get_type() == vap_type::control)
                {
                    auto meta_2_handle = std::dynamic_pointer_cast<vap_type::vap_control>(in_meta);
                    out_meta = this->handle_control_meta(meta_2_handle);
                    if(out_meta !=nullptr )
                    {
                        this->out_queue_.push(out_meta);
                        this->out_queue_semaphore_.signal();
                    }
                }   
                else
                {
                    throw std::runtime_error("invalid meta type!");
                }
                this->in_queue_.pop();
            }
        }

        virtual void dispatch_run() override 
        {
            while(true)
            {
                this->out_queue_semaphore_.wait();
                auto out_meta = out_queue_.front();
                this->push_meta(out_meta);
                this->out_queue_.pop();
            }
        }
    };

};


#endif